package offline

import org.apache.spark.ml.classification.{BinaryLogisticRegressionTrainingSummary, LogisticRegression}
import org.apache.spark.ml.feature.RFormula
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions._
import offline.SimpleFeature.feat

object LR_test {
  val spark = SparkSession
    .builder()
    .appName("LR_test")
    .enableHiveSupport()
    .getOrCreate()
  val priors = spark.sql("select * from default.priors")
  val orders = spark.sql("select * from default.orders")
  val trains = spark.sql("select * from default.trains")
  val (userFeat,prodFeat) = feat(priors,orders)
  //u1 商品集合
  //  val df_u1 = orders.filter("user_id='1'").join(priors, "order_id").select("product_id").distinct()
  //  val df_u1_train = orders.filter("user_id='1'").join(trains, "order_id").select("product_id").distinct()
  //  val intersect_products = df_u1.intersect(df_u1_train)
  val op = orders.join(priors, "order_id")
  val opTrain = orders.join(trains, "order_id")

  val user_real = opTrain.select("product_id", "user_id").distinct().withColumn("label", lit(1))
  val trainData = op.join(user_real, Seq("product_id", "user_id"), "outer")
    .select("user_id", "product_id", "label").distinct().na.fill(0)
  val train = trainData.join(userFeat, "user_id")
    .join(prodFeat, "product_id")
  val rformula = new RFormula().setFormula("label~")
  val df = rformula.fit(train).transform(train).select("features", "label")
  //实例化LR模型
  val lr = new LogisticRegression().setMaxIter(100).setRegParam(0)
  //蒋勋链接的70%作为训练 30%作为test
  val Array(trainingData, testData) = df.randomSplit(Array(0.7, 0.3))
  //训练模型
  val lrModel = lr.fit(trainingData)
  //打印参数
  println(s"weights: ${lrModel.coefficients} intercept: ${lrModel.intercept}")
  //收集日志
  val trainingSummary = lrModel.summary
  val objectHistory = trainingSummary.objectiveHistory
  //打印Loss
  objectHistory.foreach(loss => println(loss))
  //用auc来评价模型训练结果 auc是ROC曲线下的面积
  val binarySummary = trainingSummary.asInstanceOf[BinaryLogisticRegressionTrainingSummary]
  val roc = binarySummary.roc
  roc.show()
  println(binarySummary.areaUnderROC)


}
